package com.lhd.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.lhd.utils.DateFormatUtil;
import com.lhd.utils.MyKafkaUtil;
import io.debezium.data.Json;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/*
1.获取ods 中日志数topic_log
2.处理过滤topic_log
3.对新老用户进行处理
               is_news:1
                         状态=null  我要把当前人ts转换成年月日  跟新到状态值
                         状态！=null   判断当前数据的ts 和状态日期是否相同  如果不同跟新数据
               is_new:0
                         状态=null    将ts-1day 跟新到状态中
                         如果有值 不需要处理
4. 把主流数据拆分成5个流   分别存入到5个主题中
1.创建流式环境
2.设置并行度
3.运行jar 向topic_log主题 发送数据
4.从 Kafka 读取主流数据
5.数据清洗，转换结构
6.将脏数据写出到 Kafka 指定主题
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {

        // TODO 1. 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO 2. 启用状态后端


        // TODO 3. 从 Kafka 读取主流数据
        String topic = "topic_log";
        String groupId = "base_log_consumer";



        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
//        source.print();



        //page    {"commone":{},"actions":[{},{},{}],"display":[{},{},{},{}],"page":"","err":{},"ts":""}
        //start   {"commone":{},"start":{},"err":{},"ts":}
        //kafka都起来了     maxwell 运行起作用
        //hbase起不来       hbase  zk里面删除  重启
        //hive  运行内存不够

        //rich  process
       // TODO 4. 数据清洗，转换结构
        // 4.1 定义错误侧输出流
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {
        };
        SingleOutputStreamOperator<String> cleanedStream = source.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<String> out) throws Exception {
            try {


                out.collect(jsonStr);

            } catch (Exception e) {
//                            ctx.output(dirtyStreamTag, jsonStr);
                System.out.println(jsonStr);
            }
        }
    }
        );
////
////        cleanedStream.print();
////
////
////
////
//         // 4.2 将脏数据写出到 Kafka 指定主题
//        //kafka-console-produce.sh     topic_db    自己开启生产者   a,b,c
//        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyStreamTag);
////
////
////        String dirtyTopic = "dirty_data";
////        dirtyStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(dirtyTopic));
////        dirtyStream.print("脏数据:");
//        // 4.3 转换主流数据结构 jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> mappedStream = cleanedStream.map(JSON::parseObject);


        /*
        * 1.  common     page   ts
        * 2.  common    page  ts   action
        *
        * 2.  common    page  ts   dispaly
        *
        * 3. 错误  全要
        *
        * 5. 启动  不要 错误
        *
        *
        *
        * */
//        cleanedStream.map(new MapFunction<String, JSONObject>() {
//            @Override
//            public JSONObject map(String value) throws Exception {
//                return JSON.parseObject(value);
//            }
//        })
//
//
//
////      // TODO 5. 新老访客状态标记修复   状态  keyby  mid
//
//        /*
//        * 1.我们的数据按照key分组
//        * 2.给每个用户保存状态
//        * 3.首日登录的用户  1
//        *    看状态中是否有值  如果没有null  把今天的日期加进去
//        *    如果有值判断值是否相同  如果相等变成1
//        *    如果不相同   1 变成0
//        *
//        *
//        *
//        *
//        * */
//
//
//
//
//
//
//
//
//
//
////        // 5.1 按照 mid 对数据进行分组
        KeyedStream<JSONObject, String> keyedStream = mappedStream
                .keyBy(r -> r.getJSONObject("common").getString("mid"));//mid相同的给共享
//
//
//
//
//

//        // 5.2 新老访客状态标记修复   process  可以用状态  可以测输出流   还是定时器
        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                        //创建状态   给相同mid共享   容器里面只存日期   value  list   map
                    ValueState<String> firstViewDtState;


                        //open  ： 首先走open方法
                        @Override
                        public void open(Configuration param) throws Exception {
                            super.open(param);

                            firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDtState",
                                    String.class));
                        }
                        //zs   is_new=1    当前时间=dt       状态值=firstViewDt
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        // 1   0


                        //状态时间
                        String firstViewDt = firstViewDtState.value();

                        //获取当前数据的时间
                        Long ts = jsonObj.getLong("ts");


                        //把时间戳转换成年月日
                        String dt = DateFormatUtil.toDate(ts);
                        if ("1".equals(isNew)) {
                            if (firstViewDt == null) {
                                firstViewDtState.update(dt);
                            } else {

                                if (!firstViewDt.equals(dt)) {
                                    isNew = "0";
                                    jsonObj.getJSONObject("common").put("is_new", isNew);
                                }
                            }
                        } else {
                            if (firstViewDt == null) {
                                // 将首次访问日期置为昨日
                                String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                                firstViewDtState.update(yesterday);
                            }
                        }

                        out.collect(jsonObj);
                    }
                }
        );
////
//          // TODO 6. 分流 page  start     err
//        // 6.1 定义启动、曝光、动作、错误侧输出流
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };


//        /*
//       启动  common    start  ts   err
//       页面  common    page   display  action  ts  err
//        *
//        * */
/*
 {common,start,ts,error}
 {common,page,ts,action,dispaly,ts,error}

  page  大   remove actions  dispaly

  action    actions:[{}]  一个  common   page   ts


  dispaly    dispalys:[{}]  一个  common   page   ts


 */


//        // 6.2 分流
        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context context, Collector<String> out) throws Exception {
                        // 6.2.1 收集错误数据
                        JSONObject error = jsonObj.getJSONObject("err");
                        if (error != null) {
                            context.output(errorTag, jsonObj.toJSONString());
                        }
                        // 剔除 "err" 字段  {common:{},start:{},ts: , err:()}
                        jsonObj.remove("err");



                        // 6.2.2 收集启动数据
                        JSONObject start = jsonObj.getJSONObject("start");
                        if (start != null) {

                            context.output(startTag, jsonObj.toJSONString());
                        } else {
                            //page   dispaly   action
                            // 获取 "page" 字段
                            JSONObject page = jsonObj.getJSONObject("page");
                            // 获取 "common" 字段
                            JSONObject common = jsonObj.getJSONObject("common");
                            // 获取 "ts"
                            Long ts = jsonObj.getLong("ts");




                            // 6.2.3 收集曝光数据[{},{},{}]
                            JSONArray displays = jsonObj.getJSONArray("displays");
                            // [{},{},{},{}]
                            //{}
                            //{}
                            //{}
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i); //[{},{}},{]
                                    //创建空的json对象
                                    JSONObject displayObj = new JSONObject();

                                    displayObj.put("display", display);
                                    displayObj.put("common", common);
                                    displayObj.put("page", page);
                                    displayObj.put("ts", ts);

                                    context.output(displayTag, displayObj.toJSONString());
                                }
                            }
                            // 6.2.4 收集动作数据
                            JSONArray actions = jsonObj.getJSONArray("actions");
                            if (actions != null) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    JSONObject actionObj = new JSONObject();
                                    actionObj.put("action", action);
                                    actionObj.put("common", common);
                                    actionObj.put("page", page);
                                    actionObj.put("ts", ts);
                                    context.output(actionTag, actionObj.toJSONString());
                                }
                            }

                            // 6.2.5 收集页面数据
                            jsonObj.remove("displays");
                            jsonObj.remove("actions");



                            out.collect(jsonObj.toJSONString());
                        }

                    }
                }
        );
////
////        // 打印主流和各侧输出流查看分流效果
//        separatedStream.print("page>>>");
//        separatedStream.getSideOutput(startTag).print("start!!!");
//        separatedStream.getSideOutput(displayTag).print("display@@@");
//        separatedStream.getSideOutput(actionTag).print("action###");
//        separatedStream.getSideOutput(errorTag).print("error$$$");
////
////        // TODO 7. 将数据输出到 Kafka 的不同主题
////        // 7.1 提取各侧输出流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);
////        // 7.2 定义不同日志输出到 Kafka 的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
////
        separatedStream.print("page>>>>>>>>>>>>>>>>>>>");
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));

        env.execute();
    }
}
